package org.onion.pattern.event.springboot.rocketmq.publish;


import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQEventProperties;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;

import java.util.UUID;


class RocketMQProducerProvider {

    public static DefaultMQProducer createProducer(RocketMQEventProperties properties) {
        _Producer producer;
        if (StringUtils.isNotEmpty(properties.getAccessKey()) && StringUtils.isNotEmpty(properties.getSecretKey())) {
            final AclClientRPCHook aclClientRPCHook = new AclClientRPCHook(new SessionCredentials(properties.getAccessKey(), properties.getSecretKey()));
            producer = new _Producer(properties.getProducer().getGroup(), aclClientRPCHook);
        } else {
            producer = new _Producer(properties.getProducer().getGroup());
        }

        producer.setDefaultTopicQueueNums(properties.getProducer().getDefaultTopicQueueNums());
        producer.setEnabledAutoCreateTopic(properties.getProducer().isEnabledAutoCreateTopic());
        String nameServer = ""; //properties.getProducer().getNameServer();
        String namespace ="";// properties.getProducer().getNamespace();
        if (StringUtils.isEmpty(nameServer)) {
            nameServer = properties.getDefaultNameServer();
        }
        if (StringUtils.isEmpty(namespace)) {
            namespace = properties.getDefaultNamespace();
        }
        producer.setNamesrvAddr(nameServer);
        producer.setNamespace(namespace);
        producer.setInstanceName(UUID.randomUUID().toString());
        producer.setSendMsgTimeout(properties.getProducer().getSendMessageTimeout());
        producer.setCreateTopicKey(RocketMQ.AUTO_CREATE_TOPIC_KEY_TOPIC_BY_VERSION4_3);
//    producer.setDefaultTopicQueueNums(10);
        producer.setRetryTimesWhenSendFailed(properties.getProducer().getRetryTimesWhenSendFailed());
        producer.setRetryTimesWhenSendAsyncFailed(properties.getProducer().getRetryTimesWhenSendAsyncFailed());
        producer.setMaxMessageSize(properties.getProducer().getMaxMessageSize());
        producer.setCompressMsgBodyOverHowmuch(properties.getProducer().getCompressMessageBodyThreshold());
        producer.setRetryAnotherBrokerWhenNotStoreOK(properties.getProducer().isRetryNextServer());

        return producer;
    }
}
